package com.passion.mq.queues.queue1;

import com.passion.mq.config.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * 类描述：用户操作消息队列的监听器
 *
 * @author ZhuYin
 * @since 2024年12月31日
 */
@Slf4j
@Service
@ConditionalOnBean(RocketMqConfig.class)
public class QueueBatchConsumer implements MessageListenerConcurrently {

    /**
     * It is not recommend to throw exception,rather than returning ConsumeConcurrentlyStatus.RECONSUME_LATER if
     * consumption failure
     *
     * @param messages    msgs.size() >= 1<br> DefaultMQPushConsumer.consumeMessageBatchMaxSize=1,you can modify here
     * @param context
     * @return The consume status
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        try {
            // 批量处理消息
            processBatchMessages(messages);
            // 消息处理成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("Failed to process messages: {}", e.getMessage(), e);
            // 消息处理失败，稍后重试
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    private void processBatchMessages(List<MessageExt> messages) throws Exception {
        if (messages == null || messages.isEmpty()) {
            System.out.println("No messages to process.");
            return;
        }

        for (MessageExt message : messages) {
            String topic = message.getTopic();
            String messageBody = new String(message.getBody());
            String msgId = message.getMsgId();

            System.out.println("Received message from topic: " + topic + ", message: " + messageBody + ", msgId: " + msgId);

            // 处理业务逻辑
            processBusinessLogic(messageBody, msgId);
        }
    }

    private void processBusinessLogic(String messageBody, String msgId) throws Exception {
        if ("success".equals(messageBody)) {
            System.out.println("Successfully processed message with msgId: " + msgId);
        } else {
            throw new RuntimeException("Business logic failed for message with msgId: " + msgId);
        }
    }
}
